package com.baishancloud.log.common.connector.starRocks

import com.baishancloud.log.common.util.ParamUtil.paramName
import com.starrocks.connector.flink.StarRocksSink
import com.starrocks.connector.flink.table.sink.StarRocksSinkOptions
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.sink.SinkFunction

import java.util
import java.util.Properties
import java.util.function.BiConsumer
import scala.beans.BeanProperty

/**
 * 构建获取starRocks sink的参数类<br>
 * 目前只支持json字符串类型的数据写入<br>
 * 注：由于dorisDB新版本以更名为starRocks，并且连接器包只能使用starRocks的，因此本模块开发也使用starRocks命名
 *
 * @author ziqiang.wang
 * @date 2021/10/20 10:48
 */
class StarRocksSinkConnector extends Serializable {

  /**
   * 必选<br>
   * jdbc-url
   */
  @BeanProperty var jdbcUrl: String = ""
  /**
   * 必选<br>
   * load-url
   */
  @BeanProperty var loadUrl: String = ""
  /**
   * 必选<br>
   * username
   */
  @BeanProperty var username: String = ""
  /**
   * 必选<br>
   * password
   */
  @BeanProperty var password: String = ""
  /**
   * 必选<br>
   * table-name
   */
  @BeanProperty var databaseName: String = ""
  /**
   * 必选<br>
   * table-name
   *
   */
  @BeanProperty var tableName: String = ""
  /**
   * sink.properties.format<br>
   * 默认值为 json ，暂时不对外开放。
   */
  @BeanProperty var sinkPropertiesFormat: String = "json"
  /**
   * sink.properties.strip_outer_array<br>
   * 默认为true，暂时不对外开放
   *
   * @BeanProperty */
  var sinkPropertiesStripOuterArray: String = "true"
  /**
   * 用来保存其他的一些可选配置参数，可用参数如下<br>
   * sink.semantic：默认at-least-once，可用值为	at-least-once 和 exactly-once（需要开启checkpoint，在checkpoint成功之后将数据全部写入。注意，sink.buffer-flush.*类型配置都将失效）。<br><br>
   * sink.buffer-flush.max-bytes：默认94371840(90M)，累积一批序列化数据的最大大小，数值范围为：[64MB, 10GB]。<br><br>
   * sink.buffer-flush.max-rows：默认500000(50w)，累计一批数据的最大行数，数值范围为：[64,000, 5000,000]。<br><br>
   * sink.buffer-flush.interval-ms：默认300000(5m)，写入数据的时间间隔，数值范围为：[1000ms, 3600000ms]。<br><br>
   * sink.max-retries：默认1，流式load请求的最大重试次数，数值范围为：[0, 10]。<br><br>
   * sink.connect.timeout-ms：默认1000，load-url连接的超时时间，数值范围为：[100, 60000]。<br><br>
   */
  var properties: Properties = new Properties()


  /**
   * 返回该参数对象
   */
  def build(): StarRocksSinkConnector = {
    this
  }


  /**
   * 只可写入json字符串数据<br>
   */
  def sinkJsonString(): SinkFunction[String] = {
    val builder: StarRocksSinkOptions.Builder = StarRocksSinkOptions.builder()
      .withProperty("jdbc-url", jdbcUrl)
      .withProperty("load-url", loadUrl)
      .withProperty("username", username)
      .withProperty("password", password)
      .withProperty("database-name", databaseName)
      .withProperty("table-name", tableName)
      .withProperty("sink.properties.format", "json")
      .withProperty("sink.properties.strip_outer_array", "true")
    //配置其他性能调整参数`
    properties.forEach(new BiConsumer[AnyRef, AnyRef]() {
      override def accept(t: AnyRef, u: AnyRef): Unit = {
        builder.withProperty(t.toString, u.toString)
      }
    })
    StarRocksSink.sink(builder.build())
  }

}


/**
 * 构建参数对象，获取starRocks sink
 */
object StarRocksSinkConnector extends Serializable {


  /**
   * 构建方法，返回starRocks sink连尅器参数对象，后面可以通过这个对象来获取需要用到的属性<br>
   * 主参数输入值示例：--sink.jdbcUrl或者是--sink1.jdbcUrl<br>
   * 具体参数，可参考包对象
   *
   * @param parameterTool 参数工具类，包含了主方法接收到的参数
   * @param number        要获取那个数字对应的sink
   * @return
   */
  def builder(parameterTool: ParameterTool, number: String = ""): StarRocksSinkConnector = {
    val connector: StarRocksSinkConnector = new StarRocksSinkConnector()
    if (parameterTool.has(paramName(sinkJdbcUrlP, number))) {
      connector.setJdbcUrl(parameterTool.get(paramName(sinkJdbcUrlP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkJdbcUrlP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sinkLoadUrlP, number))) {
      connector.setLoadUrl(parameterTool.get(paramName(sinkLoadUrlP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkLoadUrlP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sinkUsernameP, number))) {
      connector.setUsername(parameterTool.get(paramName(sinkUsernameP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkUsernameP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sinkPasswordP, number))) {
      connector.setPassword(parameterTool.get(paramName(sinkPasswordP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkPasswordP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sinkDatabaseNameP, number))) {
      connector.setDatabaseName(parameterTool.get(paramName(sinkDatabaseNameP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkDatabaseNameP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    if (parameterTool.has(paramName(sinkTableNameP, number))) {
      connector.setTableName(parameterTool.get(paramName(sinkTableNameP, number)))
    } else {
      throw new RuntimeException(s"未设置 --${paramName(sinkTableNameP, number)} 参数，可能是在sink后面加上了数字，但是初始化时没指定数字？")
    }

    //处理其他参数
    parameterTool.toMap.entrySet().forEach((entry: util.Map.Entry[String, String]) => {
      val key: String = entry.getKey
      if (key.contains(".sink.") && key.startsWith("sink" + number)) {
        //如果这里启动输入参数包含 .sink. ，则说明设置的参数属于其他性能调整类可选参数
        connector.properties.setProperty(key.substring(key.indexOf(".") + 1), parameterTool.get(key))
      }
    })

    connector
  }


}
